package com.zyk.scaffold.tcp.disruptor;

import cn.hutool.core.thread.NamedThreadFactory;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public class DisruptorTask {

    ExecutorService executorService;
    private Disruptor<DisruptorMessage> disruptor;
    private RingBuffer<DisruptorMessage> ringBuffer;
    private LocalEventHandler disruptorMessageEventHandler;
    @PostConstruct
    public void init() throws Exception {
        log.info("starting disruptor ...");
        this.executorService = new ThreadPoolExecutor(
                Integer.valueOf(8),
                Integer.valueOf(8),
                0L,
                TimeUnit.MILLISECONDS,
                new SynchronousQueue(),
                new NamedThreadFactory("local-event-handler", true),
                (r, executor) -> {
                    try {
                        // 阻塞主线程
                        executor.getQueue().put(r);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                });
        disruptor = new Disruptor(
                LocalEventFactory.creatEventFactory(),
                1024,
                executorService,
                ProducerType.SINGLE,
                new BlockingWaitStrategy()
        );
        disruptor.handleEventsWith(LocalEventHandler.creatEventFactorys());
        disruptor.start();
        ringBuffer = disruptor.getRingBuffer();
        log.info("start disruptor finish.");
    }

    public void publishEvent(DisruptorMessage message)
    {
        long sequence = ringBuffer.next();
        try {
            //给Event填充数据
            DisruptorMessage disruptorMessage = ringBuffer.get(sequence);
            disruptorMessage.setCtx(message.getCtx());
            disruptorMessage.setMessageWrapper(message.getMessageWrapper());
            disruptorMessage.setSequence(sequence);
            log.info("往消息队列中添加消息：{}", disruptorMessage);
        } catch (Exception e) {
            log.error("failed to add event to messageModelRingBuffer for : e = {},{}",e,e.getMessage());
        } finally {
            //发布Event，激活观察者去消费，将sequence传递给改消费者
            //注意最后的publish方法必须放在finally中以确保必须得到调用；如果某个请求的sequence未被提交将会堵塞后续的发布操作或者其他的producer
            ringBuffer.publish(sequence);
        }
    }

    @PreDestroy
    public void shutdown() {
        log.info("shutdown disruptor task ...");
        // 释放disruptor资源
        disruptor.shutdown();
        disruptorMessageEventHandler.shutdown();
        log.info("shutdown disruptor task end.");
    }
}
